package com.young.tx;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;

/**
 * 事务消息生产者
 *
 * 注意：事务消息不支持延时消息和批量消息
 *
 * 事务消息的流程
 * 1、消费者发送half msg到broker
 * 2、broker通过消息发送结果回调对应方法
 *     （1）若发送成功则回调事务监听器的executeLocalTransaction 方法，执行本地事务
 *     （2）若消息发送失败，失败了就什么都不用说了，executeLocalTransaction 方法不会执行，消费者也收不到消息
 * 3、本地事务执行完毕后根据事务执行状态执行消息的commit 或者 rollback操作
 * 4、如果本地事务执行过程中出现异常没有 commit/rollback事务消息，
 *    那么broker在一定时间（这个时间可以在 broker.properties中通过`transactionMsgTimeout `属性设置）之后会调用事务监听器的checkLocalTransaction方法进行事务的回查
 * 5、根据事务的回查结果commit/rollback事务消息，如果在回查过程中出现异常无法commit/rollback事务消息，那么broker等待一定时间后会进行重试
 * 6、如果超过设置的回查次数(这个次数可以在 broker.properties中通过`transactionCheckMax`属性设置)后仍然没有commit/rollback事务消息，则broker会丢弃这个消息
 */
public class TxProducer {
    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException {
        TransactionMQProducer producer = new TransactionMQProducer("tx_producer_group");
        producer.setNamesrvAddr("192.168.188.101:9876;192.168.188.102:9876;");
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                System.out.println("执行本地事务.....");
                switch (message.getTags()) {
                    case "TagA":
                        return LocalTransactionState.COMMIT_MESSAGE;
                    case "TagB":
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    default:
                        return LocalTransactionState.UNKNOW;
                }
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println(messageExt.getTags()+"  执行事务回查.......");
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        producer.start();
        String[] tags = new String[]{"TagA", "TagB", "TagC"};
        for (int i = 0; i < 10; i++) {
            Message message = new Message("tx_topic",tags[i%tags.length], ("tx message" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            TransactionSendResult transactionSendResult = producer.sendMessageInTransaction(message, null);
            System.out.println(transactionSendResult);
        }

        //producer.shutdown();
    }
}
